package com.zhangjinghao.controller;

import com.zhangjinghao.entity.Food;
import com.zhangjinghao.es.FoodRepository;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

public class KafkaMonitor implements AcknowledgingMessageListener<String,String> {

    @Autowired
    private RedisTemplate redisTemplate;
//    @Autowired
//    private FoodRepository repository;

    @Override
    public void onMessage(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {

        System.out.println(".............................");

        String topic = consumerRecord.topic();
        String key = consumerRecord.key();
        String content = consumerRecord.value();

        System.out.println("topic is " + topic );
        System.out.println("key is " + key );
        System.out.println("content is " + content );

        // 清缓存
        redisTemplate.delete(redisTemplate.keys("*"));

//        //添加es
//        if (key.equals("insert")||key.equals("update")){
//            repository.save(content);
//        }
//        if (key.equals("del")){
//            repository.save(content);
//        }

        // 必须加上
        acknowledgment.acknowledge();
    }
}